#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# File: data_download.py
# Date: 2024/7/4
# Author: @东哥量化
# Description: 东哥量化，带你走上量化之路。


import datetime
import multiprocessing
import os
import random
import re
import akshare as ak
from tqdm import tqdm
import feather
from data_settings import *
from pyDes import des, CBC, PAD_PKCS5
import binascii
import subprocess
import platform

'''
以下内容为数据配置或公共变量
'''

# 历史数据开始时间，中国股市开始日期，上海1990年12月19日，深圳1991年7月3日。
g_start_date = '19901219'
# 历史数据结束时间（今天）
to_date = datetime.date.today().strftime('%Y%m%d')


def data_api(code, api='ak'):
    if api == 'ak':
        if period in ['daily', 'weekly', 'monthly']:

            download_df = ak.stock_zh_a_hist(symbol=code, period=period, start_date=g_start_date, end_date=to_date,
                                             adjust=adjustflag)
            df = pd.DataFrame(download_df, columns=['id', '日期', '开盘', '最高', '最低', '收盘', '成交量', '成交额'])
            df.columns = ['code', 'date', 'open', 'high', 'low', 'close', 'volume', 'amount']
            df['code'] = code
        elif period in ['1', '5', '15', '30', '60']:
            start_time = '1990-12-19 09:15:00'
            end_time = f'{str(to_date)[:4]}-{str(to_date)[5:6]}-{str(to_date)[7:]} 23:59:59'
            download_df = ak.stock_zh_a_hist_min_em(symbol=code, period=period, start_date=start_time, end_date=end_time,
                                             adjust=adjustflag)
            df = pd.DataFrame(download_df, columns=['id', "时间", "开盘", "收盘", "最高", "最低", "成交量", "成交额",])
            df.columns = ['code', 'date', 'open', 'close', 'high', 'low', 'volume', 'amount']
            df['code'] = code

    return df


def seconds_to_time(start_time, end_time):
    """
    将两个时间点之间的秒数差转换为易读的格式。

    参数:
    start_time -- 开始时间, datetime.datetime 对象。
    end_time -- 结束时间, datetime.datetime 对象。

    返回:
    时间差的字符串表示, 格式为 "DD天 HH:MM:SS"。
    """

    # 检查输入参数是否为 datetime 对象
    # 验证输入参数类型
    if not isinstance(start_time, datetime.datetime) or not isinstance(end_time, datetime.datetime):
        raise TypeError("start_time 和 end_time 必须是 datetime 对象")

    # 检查结束时间是否晚于开始时间
    # 验证时间顺序
    if end_time < start_time:
        raise ValueError("end_time 必须晚于 start_time")

    # 计算时间差
    # 计算时间差并转换为天、小时、分钟、秒的格式
    seconds = end_time - start_time  # 直接使用datetime对象相减得到timedelta对象

    # 分离时间差中的天、小时、分钟和秒
    d = seconds.days
    h = seconds.seconds // 3600
    m = (seconds.seconds // 60) % 60
    s = seconds.seconds % 60

    # 格式化时间差并返回
    # 返回格式化的时间差
    return f"{d:02d}天 {h:02d}:{m:02d}:{s:02d}"


def get_tradering_date(date=None):
    """
    取得指定日期前市场最近交易日，这里的交易日特指有日线数据的交易日

    :param date: 日期字符串，默认为空，返回历史离当日最近的有日线的交易日（当日为交易日，若17:30前调用，也无日线数据）
    :return: 最近交易日日期字符串，格式'%Y-%m-%d'
    """
    print(f'查询最近一个股票交易的数据更新日期...')
    # date为空时，将trader_date设置为当日
    if date is None:
        end_date = datetime.date.today().strftime('%Y%m%d')

    # 否则将trader_date设置为指定date
    else:
        end_date = date

    start_date = (datetime.datetime.strptime(end_date, '%Y%m%d') - datetime.timedelta(days=10)).strftime(
        '%Y%m%d')
    # 查询股票数据，query_all_stock与K线在17:30同时更新,向tradering_date前寻找有K线数据的日期
    stock_df = ak.stock_board_industry_hist_em(symbol="银行", start_date=start_date, end_date=end_date, period="日k",
                                               adjust="")

    tradering_date = datetime.datetime.strptime(stock_df['日期'].tolist()[-1], '%Y-%m-%d').strftime('%Y%m%d')
    # 获取当前时间
    now = datetime.datetime.now()
    # 获取当前日期的15时时间
    time_15 = now.replace(hour=15, minute=0, second=0, microsecond=0)
    # 判断日期是当天，并且当前时间还在交易中没有收盘
    if now < time_15 and tradering_date == datetime.date.today().strftime('%Y%m%d'):
        tradering_date = datetime.datetime.strptime(stock_df['日期'].tolist()[-2], '%Y-%m-%d').strftime('%Y%m%d')
    print(f'最近一个股票交易日期为：{tradering_date}')

    # 返回最近交易日日期
    return tradering_date


def get_stock_codes(date=None):
    """
    获取指定日期的A股代码列表

    若参数date为空，则返回最近1个交易日的A股代码列表
    若参数date不为空，且为交易日，则返回date当日的A股代码列表
    若参数date不为空，但不为交易日，则返回最近1个交易日的A股代码列表

    :param date: 日期，默认为None
    :param update: 是否更新股票列表，默认为False
    :return: A股代码的列表
    """

    # 数据库中股票代码的表名
    table_name = 'stock_codes'
    print('正在获取最新股票号码信息...')

    # 查询股票数据
    stock_list = []
    del_stock_list = []
    for mk in markets:
        if mk == 'sh':
            df = ak.stock_sh_a_spot_em()
        elif mk == 'sz':
            df = ak.stock_sz_a_spot_em()
        elif mk == 'bj':
            df = ak.stock_bj_a_spot_em()
        elif mk == 'cy':
            df = ak.stock_cy_a_spot_em()
        elif mk == 'kc':
            df = ak.stock_kc_a_spot_em()
        else:
            del_stock_list.append(mk)
        stock_list += df['代码'].tolist()
    # 如果股票列表为空，则返回错误信息，结果程序退出
    if stock_list == []:
        print('股票号码表创建失败，请检查股票市场设置有误。')
        input("程序执行完毕，请按回车键退出...")
        os._exit(0)
    # 查找退市/停牌股票
    if del_stop_stock:
        sz_delist_df = ak.stock_info_sz_delist(symbol="终止上市公司")
        sh_delist_df = ak.stock_info_sh_delist(symbol="全部")
        stop_sh = sz_delist_df['证券代码'].values.tolist()
        stop_sz = sh_delist_df['公司代码'].values.tolist()
        stop_list = stop_sh + stop_sz
        # 删除退市/停牌股票
        stock_list = [element for element in stock_list if element not in stop_list]
    if del_stock_list != []:
        for mk in del_stock_list:
            if mk == '-kc':
                # 删除科创板股票
                stock_list = list(filter(lambda x: 680000 > int(x) and int(x) > 800000, stock_list))
            elif mk == '-cy':
                # 删除科创板股票
                stock_list = list(filter(lambda x: 300000 > int(x) and int(x) > 400000, stock_list))

    stock_dict = {'code': stock_list}
    stock_df = pd.DataFrame(stock_dict)

    # 打乱股票号码,避免后面进行多线程处理是，把上市时间长的股票分配在一起，造成部分线程计算量较大，时间较长。
    random.shuffle(stock_list)

    # 检查文件夹是否存在
    if not os.path.exists(stockpath):
        # 如果文件夹不存在，创建它
        os.makedirs(stockpath)

    # 将股票代码写入数据表
    feather.write_dataframe(stock_df, f'{stockpath}/{table_name}.feather')

    print('股票号码表已创建完成。')

    # 返回股票列表

    return stock_list


def remove_negative_values(df):
    # 选择感兴趣的列
    cols_of_interest = ['open', 'high', 'close', 'low']

    # 查找每列最后一个负值的索引
    last_negative_indices = {}
    for col in cols_of_interest:
        # 查找列中所有负值的索引
        neg_indices = df[df[col] < 0].index
        # 如果存在负值，获取最后一个负值的索引
        if len(neg_indices) > 0:
            last_negative_indices[col] = neg_indices[-1]

    # 找出所有列中最后一个负值的最大索引
    if last_negative_indices:
        max_index = max(last_negative_indices.values())
    else:
        max_index = 0  # 或者设置一个默认值，比如0

    df = df.loc[max_index + 1:]

    df.reset_index(drop=True, inplace=True)

    return df


def nextday_data(code, old_df):
    """
    预测下一个交易日的股票数据。

    参数:
    code: 字符串，股票代码。
    old_df: Pandas DataFrame，历史股票数据。

    返回:
    focus_df_list: 列表，包含每个关注标签对应的DataFrame，用于记录关注的股票信息。
    """

    # 增加下一天的虚拟数据,用于计算是否有机会生成交易数据，计算结果用于加入关注数据表
    # 获取股票最后一个交易日的日期
    last_day = old_df.iloc[-1]['date']
    today = datetime.datetime.strptime(str(last_day), '%Y-%m-%d')
    # 计算明天的日期，即今天日期加上一天的时间差
    tomorrow = today + datetime.timedelta(days=1)

    if code[:2] in ['30', '68']:
        open_profit = (100 + nextday_open_20) / 100
        high_profit = (100 + nextday_high_20) / 100
        low_profit = (100 + nextday_low_20) / 100
        close_profit = (100 + nextday_close_20) / 100
    else:
        open_profit = (100 + nextday_open_10) / 100
        high_profit = (100 + nextday_high_10) / 100
        low_profit = (100 + nextday_low_10) / 100
        close_profit = (100 + nextday_close_10) / 100
    open = float(old_df.iloc[-1]['close']) * open_profit
    high = float(old_df.iloc[-1]['close']) * high_profit
    low = float(old_df.iloc[-1]['close']) * low_profit
    close = float(old_df.iloc[-1]['close']) * close_profit

    volume = float(old_df.iloc[-1]['volume'])
    amount = float(old_df.iloc[-1]['amount'])
    next_df = pd.DataFrame([[code, tomorrow, open, high, low, close, volume, amount]],
                           columns=['code', 'date', 'open', 'high', 'low', 'close', 'volume', 'amount'])
    # 将数值数据转为float型，便于后续处理
    convert_list = ['open', 'high', 'low', 'close', 'volume', 'amount']
    next_df[convert_list] = next_df[convert_list].astype(float)
    # 并接新数据
    out_df = pd.concat([old_df, next_df])
    # 重置索引
    out_df.reset_index(drop=True, inplace=True)
    old_len = len(old_df)
    # 计算扩展因子

    new_df = extend_factor(out_df, old_len)
    new_df.reset_index(drop=True, inplace=True)

    # 如果下一个交易日为买入，则把关注数据写入数据表。
    focus_df_list = []
    for tag in factorlist:
        if new_df[f'{tag}_trade'].iloc[-1] == 1:
            focus_df = old_df[-1:]
        else:
            focus_df = pd.DataFrame()
        focus_df_list.append(focus_df)

    return focus_df_list


def create_stock_data(stock_codes, mpid='单线程', update=False, tradering_date=None):
    """
    下载指定日期内，指定的指数日线数据，计算扩展因子

    :param stock_codes: 待下载数据的股票代码
    :param from_date: 日线开始日期
    :param to_date: 日线结束日期
    :param adjustflag: 复权选项 'hfq'：后复权  'qfq'：前复权  ''：不复权  默认为前复权
    :return: None
    """
    if tradering_date is None:
        tradering_date = get_tradering_date()
    else:
        tradering_date = tradering_date
    focus_df_list = []
    for tag in factorlist:
        focus_df = pd.DataFrame()
        focus_df_list.append(focus_df)
    last_df = pd.DataFrame()
    if mpid == '单线程':
        info = f'{mpid}：'
    else:
        info = f'多线程组：{mpid}，'
    for index, code in enumerate(stock_codes):
        print(f'{info}({index + 1}/{len(stock_codes)})正在创建{code}数据...')
        # 查询数据库数据
        table_name = f'{stockpath}/{code}.feather'

        try:
            if update:
                old_df = feather.read_dataframe(table_name)
                old_len = len(old_df)
            else:
                old_len = 0
        except Exception as e:
            print(f'{info}没有查询到{code}数据，将重新建立{code}数据数据表')
            old_len = 0

        out_df = data_api(api=api)

        # 剔除复权之后出现负值以及之前的所有数据
        if adjustflag != '':
            out_df = remove_negative_values(out_df)

        # 剔除停盘数据
        if out_df.shape[0]:
            out_df = out_df[(out_df['volume'] != '0') & (out_df['volume'] != '')]

        # 如果数据为空，则不创建
        if not out_df.shape[0]:
            print(f'{info}股票{code}数据为空，取消创建！！！')
            continue

        # 删除重复数据
        out_df.drop_duplicates(['date'], inplace=True)

        # 日线数据少于g_available_days_limit，则不创建
        if out_df.shape[0] < g_available_days_limit:
            print(f'{info}股票{code}长度不足，取消创建！！！')
            continue
        if len(out_df) <= old_len:
            print(f'{info}{code}已经是最新数据...')
            new_df = old_df
        else:
            if old_len != 0:
                # 校对数据
                try:
                    if old_df['date'].iloc[old_len - 1] != out_df['date'].iloc[old_len - 1] \
                            or old_df['open'].iloc[old_len - 1] != out_df['open'].iloc[old_len - 1] \
                            or old_df['open'].iloc[0] != out_df['open'].iloc[0]:
                        print(f'多线程组：{mpid}，{code}数据校对出错，重新创建{code}股票数据...')
                        old_len = 0
                except:
                    print(f'{info}{code}数据校对出错，重新创建{code}股票数据...')
                    old_len = 0
                # 如果数据和数据库数据长度相同，即没有新数据，则不创建

                if adjustflag == 'qfq':
                    if old_df['open'][0] != out_df['open'][0] or old_df['close'][0] != out_df['close'][0]:
                        print(f'{info}{code}可能进行了除权或出错，重新创建{code}股票数据...')
                        old_len = 0
                elif adjustflag == 'hfq':
                    if (old_df['open'][-1] != out_df['open'][len(old_df)-1]
                            or old_df['close'][-1] != out_df['close'][len(old_df)-1]):
                        print(f'{info}{code}可能进行了除权或出错，重新创建{code}股票数据...')
                        old_len = 0

            # 将数值数据转为float型，便于后续处理
            convert_list = ['open', 'high', 'low', 'close', 'volume', 'amount']
            out_df[convert_list] = out_df[convert_list].astype(float)
            if old_len != 0:
                # 重置索引
                out_df = pd.concat([old_df, out_df[old_len:]])
                out_df.reset_index(drop=True, inplace=True)

            # 计算扩展因子
            new_df = extend_factor(out_df, old_len)

            # 股票完整数据写入数据库
            feather.write_dataframe(new_df, table_name)
            print(f'{info}股票{code}数据更新完成！！！')
        # 检查数据是否包含当天的交易数据
        new_date = datetime.datetime.strptime(str(new_df['date'].tolist()[-1]), '%Y-%m-%d').date()
        tradering_datetime = datetime.datetime.strptime(tradering_date, '%Y%m%d').date()
        if new_date >= tradering_datetime:
            # 如果包含当天数据，则将其追加到last_df中
            if len(last_df) == 0:
                last_df = new_df[-1:]
                last_df.reset_index(drop=True, inplace=True)
            else:
                last_df = pd.concat([last_df, new_df[-1:]], ignore_index=True)
            # 生成下一个交易日的数据，并更新到对应的focus_df_list中
            new_focus_df_list = nextday_data(code, new_df)
            for i in range(len(factorlist)):
                if len(focus_df_list[i]) != 0:
                    if len(new_focus_df_list[i]) != 0:
                        focus_df_list[i] = pd.concat([focus_df_list[i], new_focus_df_list[i]], ignore_index=True)
                else:
                    if len(new_focus_df_list[i]) != 0:
                        focus_df_list[i] = new_focus_df_list[i]
        # 返回包含所有股票最新交易数据的DataFrame，以及每个关注列表的最新交易数据列表
        print(f'{info}股票{code}关注数据提取完成！！！')
    return last_df, focus_df_list


def update_stock_factor(stock_codes, mpid='单线程', tradering_date=None):
    """
    下载指定日期内，指定的指数日线数据，计算扩展因子

    :param stock_codes: 待更新因子数据的股票代码
    :param from_date: 日线开始日期
    :param to_date: 日线结束日期
    :param adjustflag: 复权选项 'hfq'：后复权  'qfq'：前复权  ''：不复权  默认为前复权
    :return: None
    """
    focus_df_list = []
    if tradering_date is None:
        tradering_date = get_tradering_date()
    else:
        tradering_date = tradering_date
    for tag in factorlist:
        focus_df = pd.DataFrame()
        focus_df_list.append(focus_df)
    last_df = pd.DataFrame()
    if mpid == '单线程':
        info = f'{mpid}：'
    else:
        info = f'多线程组：{mpid}，'
    for index, code in enumerate(stock_codes):
        print(f'{info}({index + 1}/{len(stock_codes)})正在创建{code}数据...')
        # 查询数据库数据
        table_name = f'{stockpath}/{code}.feather'
        old_df = feather.read_dataframe(table_name)

        # 计算扩展因子
        new_df = extend_factor(old_df, 0)

        # 股票完整数据写入数据库
        feather.write_dataframe(new_df, table_name)
        print(f'{info}股票{code}数据更新完成！！！')
        # 检查数据是否包含当天的交易数据
        if new_df['date'].tolist()[-1] == f'{tradering_date[:4]}-{tradering_date[4:6]}-{tradering_date[6:]}':
            # 如果包含当天数据，则将其追加到last_df中
            last_df = pd.concat([last_df, new_df[-1:]], ignore_index=True)
            # 生成下一个交易日的数据，并更新到对应的focus_df_list中
            new_focus_df_list = nextday_data(code, new_df)
            for i in range(len(factorlist)):
                focus_df_list[i] = pd.concat([focus_df_list[i], new_focus_df_list[i][-1:]], ignore_index=True)
        # 返回包含所有股票最新交易数据的DataFrame，以及每个关注列表的最新交易数据列表
    return last_df, focus_df_list


def get_code_group(process_num, stock_codes):
    """
    获取代码分组，用于多进程计算，每个进程处理一组股票

    :param process_num: 进程数
    :param stock_codes: 待处理的股票代码
    :return: 分组后的股票代码列表，列表的每个元素为一组股票代码的列表
    """

    # 创建空的分组
    code_group = [[] for i in range(process_num)]

    # 按余数为每个分组分配股票
    for index, code in enumerate(stock_codes):
        code_group[index % process_num].append(code)

    return code_group


def multiprocessing_func(func, args):
    """
    多进程调用函数

    :param func: 函数名
    :param args: func的参数，类型为元组，第0个元素为进程数，第1个元素为股票代码列表
    :return: 包含各子进程返回对象的列表
    """

    # 用于保存各子进程返回对象的列表
    async_results = []
    final_results = []
    argslist = list(args)
    # 创建进程池
    with multiprocessing.Pool(processes=args[0]) as pool:
        # 多进程异步计算
        mpid = 0
        for codes in get_code_group(args[0], args[1]):
            mpid += 1
            argslist[0] = codes
            argslist[1] = mpid
            async_result = pool.apply_async(func, args=tuple(argslist))
            async_results.append(async_result)

        # 阻止后续任务提交到进程池
        pool.close()

        # 等待所有进程结束
        pool.join()

        # 获取并处理每个子进程的结果
        for result in async_results:
            try:
                # 使用get(timeout=None)等待直到结果可用，这会阻塞直到结果准备好
                final_results.append(result.get(timeout=None))
            except Exception as e:
                # 记录或处理异常，确保一个任务的失败不影响其他任务
                final_results.append(f"Error in subprocess: {str(e)}")

    return final_results


def create_last_focus_data(results, tradering_date, check=False):
    """
    根据计算结果创建并更新关注股票的数据框架。

    :param results: 计算结果的列表，每个结果是一个包含最后交易日数据和关注股票数据的元组。
    :param tradering_date: 交易日期，用于为数据框架文件名添加日期后缀。
    :param check: 布尔值，指示是否检查并更新现有的数据框架文件。
    :return: 如果创建的数据框架为空，则返回None；否则返回更新后的最后交易日数据框架。
    """
    # 初始化最后交易日数据框架
    last_df = pd.DataFrame()
    # 初始化关注股票的数据框架列表
    focus_df_list = []
    for tag in factorlist:
        focus_df_list.append(pd.DataFrame())
    # 遍历交易结果，合并最后交易日数据和关注股票数据
    for result in results:
        result_last_df = result[0]
        result_focus_df_list = result[1]
        # 如果最后交易日数据为空，则跳过
        if len(result_last_df) == 0:
            continue
        # 合并最后交易日数据
        if len(last_df) == 0:
            last_df = result_last_df
            last_df.reset_index(drop=True, inplace=True)
        else:
            if len(result_last_df) != 0:
                last_df = pd.concat([last_df, result_last_df], ignore_index=True)
        for i in range(len(factorlist)):
            if len(focus_df_list[i]) != 0:
                if len(result_focus_df_list[i]) != 0:
                    focus_df_list[i] = pd.concat([focus_df_list[i], result_focus_df_list[i]], ignore_index=True)
            else:
                if len(result_focus_df_list[i]) != 0:
                    focus_df_list[i] = result_focus_df_list[i]
    # 如果最后交易日数据框架为空，则返回None
    if len(last_df) == 0:
        return None
    # 如果检查参数为True，更新现有的数据框架文件
    if check:
        # 读取并更新最后交易日数据框架文件
        old_df = feather.read_dataframe(f'{stockpath}/stock_last.feather')
        new_df = pd.concat([old_df, last_df], ignore_index=True)
        feather.write_dataframe(new_df, f'{stockpath}/stock_last.feather')
        feather.write_dataframe(new_df, f'{stockpath}/stock_last_{tradering_date}.feather')
        # 读取并更新最后交易日数据框架文件
        for i in range(len(focus_df_list)):
            old_df = feather.read_dataframe(f'{stockpath}/stock_focus_{factorlist[i]}.feather')
            focus_df = pd.concat([old_df, focus_df_list[i]], ignore_index=True)
            feather.write_dataframe(focus_df, f'{stockpath}/stock_focus_{factorlist[i]}.feather')
            feather.write_dataframe(focus_df, f'{stockpath}/stock_focus_{factorlist[i]}_{tradering_date}.feather')
    # 如果检查参数为False，直接写入新的数据框架文件
    else:
        feather.write_dataframe(last_df, f'{stockpath}/stock_last.feather')
        feather.write_dataframe(last_df, f'{stockpath}/stock_last_{tradering_date}.feather')
        for i in range(len(focus_df_list)):
            feather.write_dataframe(focus_df_list[i], f'{stockpath}/stock_focus_{factorlist[i]}.feather')
            feather.write_dataframe(focus_df_list[i],
                                    f'{stockpath}/stock_focus_{factorlist[i]}_{tradering_date}.feather')


def create_data_mp(stock_codes, tradering_date, update=False, check=False):
    """
    使用多进程方式创建股票数据。

    该函数通过多进程并行化创建指定股票代码列表中每个股票的日线数据，并计算扩展因子。
    如果更新参数为True，则会尝试更新已有数据；如果检查参数为True，则会在创建数据后进行数据完整性检查。

    :param stock_codes: 股票代码列表，用于指定需要创建数据的股票。
    :param update: 布尔值，指定是否更新已有数据，默认为False。
    :param check: 布尔值，指定是否检查数据完整性，默认为False。
    :return: 无返回值。
    """

    # 打印开始下载数据的提示信息。
    print(f'开始下载股票交易数据，并根据股票号码创建数据表...')
    # 使用多进程函数处理股票数据创建任务。
    results = multiprocessing_func(create_stock_data, (process_num, stock_codes, update, tradering_date,))
    # 根据处理结果和当前交易日，创建或更新最后关注的数据。
    print(f'根据股票号码创建数据表完成，开始计算last_focus数据...')
    create_last_focus_data(results, tradering_date, check=check)


def factor_updata_mp(stock_codes):
    """
    多进程更新股票因子数据。

    通过多进程方式调用update_stock_factor函数，更新指定股票代码列表的因子数据，并创建最新的关注股票数据。

    参数:
    stock_codes: list, 股票代码列表，用于更新因子数据。

    返回:
    无直接返回值，但会生成更新后的股票因子数据和最新的关注股票数据。
    """
    # 获取当前交易日日期，用于数据更新和创建最新关注股票数据
    # 获取最新交易日
    tradering_date = get_tradering_date()

    # 打印开始更新股票因子数据的提示信息
    print(f'根据股票号码开始重新计算股票因子数据，并创建数据表...')

    # 使用多进程方式更新股票因子数据，传入进程数、股票代码列表和交易日日期
    # 使用多进程函数处理股票数据创建任务。
    results = multiprocessing_func(update_stock_factor, (process_num, stock_codes, tradering_date,))

    # 根据更新结果，创建或更新最新的关注股票数据
    # 根据处理结果和当前交易日，创建或更新最后关注的数据。
    create_last_focus_data(results, tradering_date)


def extend_factor(df, oldindex):
    """
    计算扩展因子
    pt是通用因子计算函数，分别进行基础因子计算和某种策略处理，
    oldindex为原始股票列表的索引用于更新时不对旧数据再次计算，减少计算次数提高运算效率。
    'aa',为策略'aatrade'的扩展因子计算方法的缩写，以区分不同的策略。
    profit=4，为策略bbtrade的利润百分比。如果策略为固定的利润百分比，即买入后的某个交易日盘中达到10%就卖，
    那么profit参数可以设置为固定值，如profit=10，表示利润百分比为10%。


    :param df: 待计算扩展因子的DataFrame
    :param oldindex: 旧的索引
    :param rounding_precision: 四舍五入的精度，默认为4
    :return: 包含扩展因子的DataFrame
    """

    processed_df = df.copy()

    processed_df = processed_df.pipe(basic_factors, oldindex)

    processed_df.fillna(0, inplace=True)

    for factor in factorlist:

        if factor in list(profit_dict.keys()):

            profit = int(profit_dict[factor])
        else:
            profit = 0

        processed_df = processed_df.pipe(process_tradedata, oldindex, factor, f'{factor}_buy2sell', profit=profit)

    # 下面内容保留，不要删除
    # 将DataFrame中的NaN值填充为0
    processed_df.fillna(0, inplace=True)

    # 将DataFrame中的数值四舍五入到小数点后4位,可修改数字为自己需要的精度
    processed_df = processed_df.round(4)

    # processed_df = processed_df[columns]

    return processed_df


# 使用函数名调用函数
def call_function_by_name(func_name, *args, **kwargs):
    """
    根据函数名动态调用函数。

    本函数通过函数名查找并调用全局范围内的函数，并传递相应的参数。
    这种做法提供了动态调用功能的灵活性，可以在运行时根据需要决定调用哪个函数。

    参数:
    - func_name: 字符串，表示要调用的函数的名称。
    - *args: 可变参数，传递给目标函数的参数列表。
    - **kwargs: 可变关键字参数，传递给目标函数的关键字参数字典。

    返回:
    - 调用目标函数后的返回值。

    注意:
    - 本函数依赖于globals()函数来查找函数，要求被调用的函数必须在全局范围内定义。
    - 如果函数名不存在或不是可调用的，则不会发生函数调用，本函数直接返回None。
    """

    func = globals().get(func_name)
    if callable(func):
        return func(*args, **kwargs)


def process_tradedata(df, oldindex, Label, trade_func, profit=0):
    """
    处理交易数据，根据交易策略标记买卖点及计算相关指标。

    :param df: 包含交易数据的DataFrame，将在这个DataFrame上进行操作。
    :param oldindex: 开始处理数据的索引位置。
    :param Label: 用于标记买卖点和相关指标的前缀标签。
    :param trade_func: 交易策略函数，根据当前市场状况决定是否买卖。
    :param profit: 固定利润目标，用于提前平仓。
    :return: 新的DataFrame，包含买卖点标记和相关交易指标。
    """
    # 从oldindex开始遍历DataFrame
    for i in range(oldindex, len(df)):
        # 初始化买卖点及相关指标的值
        if i <= 10:
            init_values = [f'{Label}_b2s', f'{Label}_b2smarket', f'{Label}_trade', f'{Label}_buyprice',
                           f'{Label}_profit', f'{Label}_profits', f'{Label}_holddate', f'{Label}_trades',
                           f'{Label}_win']
            df.loc[i, init_values] = 0
        else:

            # 根据是否有固定利润目标，调用不同的交易策略函数
            # 调用相应的交易函数
            if profit == 0:
                buy2sell = call_function_by_name(trade_func, df, i)
            else:
                buy2sell = call_function_by_name(trade_func, df, i, profit)
            # 标记买卖点
            df.at[i, f'{Label}_b2s'] = buy2sell
            # 根据买卖点更新市场状态，买方市场还是卖方市场
            if df[f'{Label}_b2s'][i] == 0:
                df.at[i, f'{Label}_b2smarket'] = df[f'{Label}_b2smarket'][i - 1]
            else:
                df.at[i, f'{Label}_b2smarket'] = df[f'{Label}_b2s'][i]
            # 根据市场买卖点标记更新交易者标记，只留下买入/卖出点信号，其它为0
            if df[f'{Label}_b2smarket'][i] == df[f'{Label}_b2smarket'][i - 1]:
                df.at[i, f'{Label}_trade'] = 0
            else:
                df.at[i, f'{Label}_trade'] = df[f'{Label}_b2smarket'][i]
            # 根据买卖信号更新买入价格、持有天数、利润等相关指标
            if df[f'{Label}_trade'][i] == 1:
                # 买入操作，更新数据
                df.at[i, f'{Label}_buyprice'] = df['close'][i]
                df.at[i, f'{Label}_holddate'] = 1
                df.at[i, f'{Label}_profit'] = 0
                df.at[i, f'{Label}_profits'] = df[f'{Label}_profits'][i - 1]
                df.at[i, f'{Label}_trades'] = df[f'{Label}_trades'][i - 1]
            elif df[f'{Label}_trade'][i] == -1 and df[f'{Label}_holddate'][i - 1] != 0:
                # 卖出操作，更新数据
                df.at[i, f'{Label}_buyprice'] = df[f'{Label}_buyprice'][i - 1]
                df.at[i, f'{Label}_holddate'] = 0
                if profit == 0:
                    # 没有固定利润目标时，根据收盘价计算利润
                    df.at[i, f'{Label}_profit'] = round(
                        (df['close'][i] - df[f'{Label}_buyprice'][i]) / df[f'{Label}_buyprice'][i] * 100, 4)
                else:
                    # 有固定利润目标时，根据最高价判断是否提前平仓
                    if df['high'][i] > df[f'{Label}_buyprice'][i - 1] * (1 + float(profit) / 100):
                        df.at[i, f'{Label}_profit'] = float(profit)
                    else:
                        df.at[i, f'{Label}_profit'] = round(
                            (df['close'][i] - df[f'{Label}_buyprice'][i]) / df[f'{Label}_buyprice'][i] * 100, 4)
                df.at[i, f'{Label}_profits'] = round(df[f'{Label}_profits'][i - 1] + df[f'{Label}_profit'][i], 4)
                df.at[i, f'{Label}_trades'] = round(df[f'{Label}_trades'][i - 1] + 1, 4)
            else:
                # 没有买卖信号时，更新买入价格、持有天数、利润等相关指标
                if df[f'{Label}_trade'][i - 1] == -1:
                    df.at[i, f'{Label}_buyprice'] = 0
                else:
                    df.at[i, f'{Label}_buyprice'] = df[f'{Label}_buyprice'][i - 1]
                if df[f'{Label}_holddate'][i - 1] == 0:
                    df.at[i, f'{Label}_holddate'] = 0
                else:
                    df.at[i, f'{Label}_holddate'] = df[f'{Label}_holddate'][i - 1] + 1
                df.at[i, f'{Label}_profit'] = 0
                df.at[i, f'{Label}_profits'] = df[f'{Label}_profits'][i - 1]
                df.at[i, f'{Label}_trades'] = df[f'{Label}_trades'][i - 1]
            # 根据利润更新盈利次数
            if df[f'{Label}_profit'][i] > 0:
                df.at[i, f'{Label}_win'] = df[f'{Label}_win'][i - 1] + 1
            else:
                df.at[i, f'{Label}_win'] = df[f'{Label}_win'][i - 1]

    # 计算并添加成功率和收益率指标
    new_df = df.copy()
    new_df[f'{Label}_success'] = round(new_df[f'{Label}_win'] / new_df[f'{Label}_trades'] * 100, 4)

    win_avg = (new_df[f'{Label}_profit'].where(new_df[f'{Label}_profit'] > 0).expanding().mean().ffill())
    new_df[f'{Label}_winearnings'] = win_avg
    lose_avg = (new_df[f'{Label}_profit'].where(new_df[f'{Label}_profit'] < 0).expanding().mean().ffill())
    new_df[f'{Label}_loseearnings'] = lose_avg
    new_df[f'{Label}_earnings'] = round((new_df[f'{Label}_winearnings'] / 100 * new_df[f'{Label}_success'] / 100 +
                                         new_df[f'{Label}_loseearnings'] / 100 * (
                                                     1 - new_df[f'{Label}_success'] / 100)) * 100, 4)

    new_df[f'{Label}_maxprofit'] = new_df[f'{Label}_profit'].expanding().max()
    new_df[f'{Label}_maxdrawdown'] = new_df[f'{Label}_profit'].expanding().min()
    new_df[f'{Label}_winstreak'] = ((new_df[f'{Label}_profit'] > 0).astype(int).where(lambda x: x == 1, 0)
                                    .groupby((new_df[f'{Label}_profit'] >= 0).astype(int).ne(1).cumsum())
                                    .cumsum()).expanding().max()
    new_df[f'{Label}_losestreak'] = ((new_df[f'{Label}_profit'] < 0).astype(int).where(lambda x: x == 1, 0)
                                     .groupby((new_df[f'{Label}_profit'] <= 0).astype(int).ne(1).cumsum())
                                     .cumsum()).expanding().max()

    new_df[f'{Label}_profit_temp'] = (df[f'{Label}_profit'] / 100 + 1)
    for i in range(len(new_df)):
        if i == 0:
            new_df.at[i, f'{Label}_streak_temp'] = 1
        elif ((new_df[f'{Label}_profit_temp'][i] >= 1 and new_df[f'{Label}_streak_temp'][i - 1] >= 1)
              or (new_df[f'{Label}_profit_temp'][i] <= 1 and new_df[f'{Label}_streak_temp'][i - 1] <= 1)):
            new_df.at[i, f'{Label}_streak_temp'] = new_df[f'{Label}_profit_temp'][i] * new_df[f'{Label}_streak_temp'][
                i - 1]
        else:
            new_df.at[i, f'{Label}_streak_temp'] = new_df[f'{Label}_profit_temp'][i]
    new_df[f'{Label}_profitstreak'] = new_df[f'{Label}_streak_temp'].expanding().max()
    new_df[f'{Label}_drawdownstreak'] = new_df[f'{Label}_streak_temp'].expanding().min()

    return new_df


def get_filenames_in_directory(directory_path):
    """
    获取指定目录下的所有文件名。

    :param directory_path: 文件夹的路径 (字符串)
    :return: 包含文件名的列表 (列表)
    """
    # 确保传入的路径是绝对路径，以避免相对路径可能引起的错误
    directory_path = os.path.abspath(directory_path)

    # 检查路径是否存在且是一个目录
    if os.path.exists(directory_path) and os.path.isdir(directory_path):
        # 使用os.listdir获取目录下的所有条目
        entries = os.listdir(directory_path)

        # 过滤出文件名，排除可能的子目录
        filenames = [entry for entry in entries if os.path.isfile(os.path.join(directory_path, entry))]
        table_list = [name.replace('.feather', '') for name in filenames if
                      re.match(r'^\d+$', name.replace('.feather', ''))]
        return table_list
    else:
        print(f"路径不存在或不是一个有效的目录: {directory_path}")
        return []


def update_check(stock_codes=None, tradering_date=None):
    print('更新完成，正在检查股票更新是否存在缺漏！')

    if stock_codes is None:
        # 获取所有股票数据表名称
        stock_codes = get_filenames_in_directory(stockpath)
    if tradering_date is None:
        # 获取最新交易日
        tradering_date = get_tradering_date()
    # 获取最新交易日
    stock_list = []
    for code in tqdm(stock_codes):
        try:
            trader_df = feather.read_dataframe(f'{stockpath}/{code}.feather')
            if len(trader_df) == 0:
                stock_list.append(code)
        except:
            stock_list.append(code)

    if len(stock_list) == 0:
        print('股票更新不存在缺漏，更新完成。')

    elif len(stock_list) < process_num:
        print(f'目前共有{len(stock_list)}只股票数据更新存在缺漏，正在对缺漏的股票进行数据更新中...')
        last_df, focus_df_list = create_stock_data(stock_list, mpid='单线程', update=True)  # 单线程
        if len(last_df) == 0:
            print('股票更新完成，正在检查股票更新是否存在缺漏！')
        else:
            old_last_df = feather.read_dataframe(f'{stockpath}/stock_last.feather')
            new_last_df = pd.concat([old_last_df, last_df], ignore_index=True)
            feather.write_dataframe(new_last_df, f'{stockpath}/stock_last.feather')
            for i in range(len(focus_df_list)):
                old_focus_df = feather.read_dataframe(f'{stockpath}/stock_focus_{factorlist[i]}.feather')
                focus_df_list[i] = pd.concat([old_focus_df, focus_df_list[i]], ignore_index=True)
                feather.write_dataframe(focus_df_list[i], f'{stockpath}/stock_focus_{factorlist[i]}.feather')

    else:
        print(f'目前共有{len(stock_list)}只股票数据更新存在缺漏，正在对缺漏的股票进行数据更新中...')
        create_data_mp(stock_list, tradering_date, update=True, check=True)  # 多线程


def repair_focus(stock_codes, mpid, tradering_date):
    """
    修复交易关注数据表的函数。

    使用多进程来处理股票代码列表，针对每个股票代码，读取其交易数据，
    并更新最新的交易数据到对应的关注数据表中。

    参数:
    stock_codes: 股票代码列表，用于指定需要处理的股票范围。
    mpid: 多进程标识符，用于区分不同的进程。
    tradering_date: 交易日期字符串，格式为'YYYYMMDD'，用于筛选最新的交易数据。

    返回:
    last_df: 包含所有股票最新交易数据的DataFrame。
    focus_df_list: 一个列表，其中每个元素是一个DataFrame，用于存储每个关注列表的最新交易数据。
    """
    """
    使用多进程修复交易关注数据表
    """
    # 初始化一个空的DataFrame，用于存储所有股票的最新交易数据
    last_df = pd.DataFrame()
    # 初始化一个列表，用于存储每个关注列表的最新交易数据
    focus_df_list = []
    for tag in factorlist:
        focus_df_list.append(pd.DataFrame())
    for index, code in enumerate(stock_codes):
        # 打印进度信息，方便用户了解当前处理的股票代码
        print(f'多线程组（{mpid}）：({index + 1}/{len(stock_codes)})正在读取{code}数据...')
        try:
            # 读取指定股票代码的交易数据
            df = feather.read_dataframe(f'{stockpath}/{code}.feather')
        except:
            continue
        # 检查数据是否包含当天的交易数据
        if df['date'].tolist()[-1] == f'{tradering_date[:4]}-{tradering_date[4:6]}-{tradering_date[6:]}':
            # 如果包含当天数据，则将其追加到last_df中
            last_df = pd.concat([last_df, df[-1:]], ignore_index=True)
            # 生成下一个交易日的数据，并更新到对应的focus_df_list中
            new_focus_df_list = nextday_data(code, df)
            for i in range(len(factorlist)):
                focus_df_list[i] = pd.concat([focus_df_list[i], new_focus_df_list[i][-1:]], ignore_index=True)
    # 返回包含所有股票最新交易数据的DataFrame，以及每个关注列表的最新交易数据列表
    return last_df, focus_df_list


def repair_focus_mp(stock_codes, tradering_date):
    """
    使用多进程修复交易关注数据表

    :param stock_codes: 待创建数据的股票代码
    :param process_num: 进程数
    :return: None
    """

    # 调用多进程任务
    results = multiprocessing_func(repair_focus, (process_num, stock_codes, tradering_date,))
    create_last_focus_data(results, tradering_date)


def stock_updata_all(update=False):
    """
    主函数，负责股票数据的更新。

    参数:
    update -- 布尔类型，指示是否更新已有数据，默认为False。

    主要功能包括：
    1. 获取股票代码和交易日期。
    2. 使用多线程创建数据地图。
    3. 检查并更新数据的完整性。
    4. 打印程序的运行时间。
    """

    # 记录程序开始时间
    start_time = datetime.datetime.now()

    # 获取最新交易日
    tradering_date = get_tradering_date()

    # 获取股票代码
    stock_codes = get_stock_codes()

    # 创建股票数据，可以指定是否更新/覆盖已有数据
    create_data_mp(stock_codes, tradering_date, update=update)  # 多线程

    # 检查并更新数据完整性
    update_check(stock_codes, tradering_date)  # 升级完整性检查

    # 记录程序结束时间
    end_time = datetime.datetime.now()

    # 打印程序运行时间
    print(f'程序运行时间：{seconds_to_time(start_time, end_time)}')


def stock_updata_factor():
    """
    更新股票因子数据。

    该函数定期执行，用于从源数据中提取股票因子，并更新到相应的数据表中。
    """
    # 记录程序开始时间
    start_time = datetime.datetime.now()

    # 获取目录中所有股票代码文件名
    # 获取所有股票数据表名称
    stock_codes = get_filenames_in_directory(stockpath)
    # 使用多进程更新股票因子数据
    factor_updata_mp(stock_codes)

    # 记录程序结束时间
    end_time = datetime.datetime.now()
    # 打印程序运行时间
    print(f'程序运行时间：{seconds_to_time(start_time, end_time)}')


def stock_updata_main(mode='all'):
    if mode == 'all':
        # 覆盖式下载股票数据及股票因子计算
        stock_updata_all(update=False)
    if mode == 'update':
        # 更新式下载股票数据及股票因子计算
        stock_updata_all(update=True)
    elif mode == 'factor':
        # 使用原有股票数据更新股票因子计算，适用于在股票数据已经为最新的情况下，调试/重置因子
        stock_updata_factor()
    else:
        print('mode参数错误! 请检查参数输入是否有误，确认后重新运行程序！')


def stock_updata_test(stock_num=0, stock_codes=[]):
    """
    主函数，负责股票数据的更新。

    参数:
    update -- 布尔类型，指示是否更新已有数据，默认为False。

    主要功能包括：
    1. 获取股票代码和交易日期。
    2. 使用多线程创建数据地图。
    3. 检查并更新数据的完整性。
    4. 打印程序的运行时间。
    """

    # 记录程序开始时间
    start_time = datetime.datetime.now()

    # 获取最新交易日
    tradering_date = get_tradering_date()

    # 获取股票代码
    if stock_codes == []:
        stock_codes = get_stock_codes()
    if stock_num != 0:
        stock_codes = stock_codes[:stock_num]

    # 创建股票数据，可以指定是否更新/覆盖已有数据
    create_data_mp(stock_codes, tradering_date, update=False)  # 多线程

    # 检查并更新数据完整性
    update_check(stock_codes, tradering_date)  # 升级完整性检查

    # 记录程序结束时间
    end_time = datetime.datetime.now()

    # 打印程序运行时间
    print(f'程序运行时间：{seconds_to_time(start_time, end_time)}')


def get_os_name():
    # 获取操作系统名称
    os_name = platform.system()
    if os_name == "Windows":
        try:
            # 获取CPU序列号（注意：并非所有系统都提供CPU序列号）
            result = subprocess.run(['wmic', 'cpu', 'get', 'ProcessorId', '/value'], stdout=subprocess.PIPE, text=True)
            if result.returncode == 0:
                # 清理输出，只保留需要的部分
                for line in result.stdout.splitlines():
                    if 'ProcessorId=' in line:
                        return line.split('=')[-1].strip()
            else:
                return 'WinPxdJjT3Babx1r'
        except Exception as e:
            return 'WinDTsh1dDocnG9s'
    elif os_name == "Linux":
        try:
            result = subprocess.run(['lscpu'], stdout=subprocess.PIPE, text=True)
            if result.returncode == 0:
                return result.stdout
                # 你可能需要根据输出解析特定的信息
            else:
                return 'linT3BlbkFJubLcn'
        except Exception as e:
            return 'linqPxdJbx1BlbYM'
    else:
        return "Other3wXHdDoqoWl"

